package com.offcn.bigdata.spark.p2

import com.offcn.bigdata.spark.p2._02CombineByKeyOps._
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * aggregateByKey和combineByKey的区别相当于foldByKey和reduceByKey
  */
object _03AggregateBykeyOps {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
            .setMaster("local[*]")
            .setAppName(s"${_03AggregateBykeyOps.getClass.getSimpleName}")
        val sc = new SparkContext(conf)
        val infos = sc.parallelize(List(
            Student("郭雪磊", 18, "山东"),
            Student("单 松", 20, "山东"),
            Student("刘宇航", 18, "河北"),
            Student("王健", 18, "河南"),
            Student("许迎港", 18, "河北"),
            Student("元永劫", 18, "黑龙江"),
            Student("林博", 18, "黑龙江"),
            Student("李佳奥", 18, "河南"),
            Student("冯世明", 18, "黑龙江"),
            Student("肖楚轩", 18, "山东"),
            Student("张皓", 18, "河南"),
            Student("冯岩", 18, "黑龙江")
        ), 2)
//        abk2gbk(infos)
        abk2rbk(infos)

        sc.stop()
    }

    def abk2rbk(infos: RDD[Student]): Unit = {
        val p2Info = infos.mapPartitionsWithIndex((index, it) => {
            val list = it.toList
            println(s"分区编号{${index}}中的数据为： ${list}")
            list.map(stu => (stu.province, 1)).toIterator
        })

        p2Info.aggregateByKey(0)(_+_, _+_).foreach{case (province, num) => {
            println(s"省份：${province}对应的人数为：${num}")
        }}
    }

    def abk2gbk(infos: RDD[Student]): Unit = {
        val p2Info = infos.mapPartitionsWithIndex((index, it) => {
            val list = it.toList
            println(s"分区编号{${index}}中的数据为： ${list}")
            list.map(stu => (stu.province, stu)).toIterator
        })

        p2Info.aggregateByKey(Array[Student]())(
            (array: Array[Student], stu: Student) => array.+:(stu),
            (array1: Array[Student], array2: Array[Student]) => array1 ++ array2
        ).foreach{case (province, infos) => {
            println(s"省份：$province, 信息：${infos.toList}")
        }}
    }
}
